package com.atguigu.chapter05.Transform;

import com.atguigu.chapter05.WaterSensor;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/3 15:29
 */
public class Flink02_RichFunction {
    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);

        DataStreamSource<String> inputDS = env
//                .readTextFile("input/sensor.csv");
                .socketTextStream("localhost",9999);

        // TODO Transform  -  富函数

        SingleOutputStreamOperator<WaterSensor> sensorDS = inputDS.map(new MyRichMap());

        sensorDS.print();

        env.execute();
    }

    public static class MyRichMap extends RichMapFunction<String,WaterSensor>{


        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("open....");
        }

        @Override
        public void close() throws Exception {
            System.out.println("close....");
        }

        @Override
        public WaterSensor map(String value) throws Exception {
            RuntimeContext runtimeContext = getRuntimeContext();
            System.out.println(
                    runtimeContext.getTaskName() + "\n" +
                    runtimeContext.getTaskNameWithSubtasks()
            );
            String[] line = value.split(",");
            return new WaterSensor(line[0], Long.valueOf(line[1]), Integer.valueOf(line[2]));
        }
    }
}
/**
 * 富函数：
 *  1）生命周期管理方法： open()、close()
 *      注意：如果是读文件，每个 并行实例 都会 多执行一次 close()
 *  2）运行时上下文： RuntimeContext => 获取 环境信息、状态(累加器、计数器.........很多很多)
 */